package com.example.pro2_reactive_streams;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.TimeUnit;

/**
 * @program: reactivePro
 * @description:  用 reactive stream 的Publisher、Processor、Subscriber 重新实现汽车生产流程
 * 汽车生产流程: 安装车门, 安装轮子, 安装座椅
 * 需求: 安装车门, 安装轮子, 安装座椅, 耗时 2 秒, 1.5 秒, 2.5 秒
 * @author: zy
 * @create: 2025-04-23 15:24
 */
public class Pro2_Publisher_Processor_Subscriber {

    // 使用线程池来执行耗时任务，实现异步非阻塞
    private static final ExecutorService executor = Executors.newSingleThreadExecutor();

    public static void main(String[] args) throws Exception {
        makeCar7();
        System.out.println("等待线程池任务完成...");

        executor.awaitTermination(10, TimeUnit.MINUTES);
        // 关闭线程池，并等待任务完成，最多等待 10 分钟
        executor.shutdown();
        System.out.println("程序完成。");
    }

    // 安装车门, 耗时 2 秒
    static String installDoor(String car) {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("安装车门");
        return car + "+车门";
    }
    // 安装轮子, 耗时 1.5 秒
    static String installWheel(String car) {
        try {
            Thread.sleep(1500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("安装轮子");
        return car + "+轮子";
    }
    // 安装座椅, 耗时 2.5 秒
    static String installSeat(String car) {
        try {
            Thread.sleep(2500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("安装座椅");
        return car + "+座椅";
    }


    static void makeCar7() {
        // 采用的方案是 reactive stream方案.    publisher  (subsciption), operator,   (subsciption)  subscriber
        // 发布者：　原始数据　，　定义subscription
        Flow.Publisher<String> pub = new Flow.Publisher<String>() {
            private Flow.Subscriber<? super String> subscriber;  //下一级订阅者

            @Override
            public void subscribe(Flow.Subscriber<? super String> subscriber) {
                this.subscriber = subscriber;
                // 订阅媒介 subscription
                subscriber.onSubscribe(new Flow.Subscription() {
                    String car = "汽车";  //初始数据
                    int index;   //经过几次处理
                    boolean cancel;
                    @Override
                    public void request(long n) {  //背压  ,表示请求几条数据
                        for (int i = 0; i < n; i++) {
                            if (cancel) {
                                return;
                            }
                            // 发送 4 个数据之后, 中断执行
                            if (index >= n) {
                                subscriber.onComplete();   // 完成数据的获取
                                cancel = true;
                                return;
                            }
                            // 触发订阅者的 onNext
                            // 异步发送数据
                            executor.submit(() -> {
                                subscriber.onNext(car + (++index));
                            });
                        }
                    }

                    @Override
                    public void cancel() {
                        cancel = true;
                        System.out.println("Publisher subscription canceled.");
                    }
                });
            }
        };


        // 处理者(中间步骤: 安装车门)
        Flow.Processor<String, String> installDoor = new Flow.Processor<>() {
            private Flow.Subscriber<? super String> subscriber;  //下一级订阅者
            private Flow.Subscription subscription;

            @Override
            public void subscribe(Flow.Subscriber<? super String> subscriber) {
                this.subscriber = subscriber;
            }

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                // 下一级订阅者订阅
                subscriber.onSubscribe(subscription);
            }
            @Override  //上一级订阅者数据来了
            public void onNext(String item) {
                executor.submit(() -> {
                    //  执行当前步骤方法
                    String newitem = installDoor(item);
                    // 触发下一个订阅者 onNext
                    subscriber.onNext(newitem);
                });
            }
            @Override
            public void onError(Throwable throwable) {
                subscriber.onError(throwable);
            }
            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        };


        // 处理者(中间步骤: 安装轮子)
        Flow.Processor<String, String> installWheel = new Flow.Processor<>() {
            private Flow.Subscriber<? super String> subscriber;
            private Flow.Subscription subscription;

            @Override
            public void subscribe(Flow.Subscriber<? super String> subscriber) {
                this.subscriber = subscriber;
            }

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscriber.onSubscribe(subscription);
            }

            @Override
            public void onNext(String item) {
                executor.submit(() -> {
                    String newItem = installWheel(item);
                    subscriber.onNext(newItem);
                });
            }

            @Override
            public void onError(Throwable throwable) {
                subscriber.onError(throwable);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        };


        // 订阅者(最终步骤: 安装座位)
        Flow.Subscriber<String> installSeat = new Flow.Subscriber<String>() {
            // 订阅媒介
            private Flow.Subscription subscription;
            private int receivedCount = 0; // 接收数据计数

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                // 首次请求
                subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                executor.submit(() -> {
                    String newItem = installSeat(item);
                    System.out.println("makeCar7 完成! " + newItem);
                    // 完成一个数据处理后，再次请求一个
                    subscription.request(2);
                });
            }

            @Override
            public void onError(Throwable throwable) {
                System.err.println("Error: " + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("installSeat subscriber onComplete: 所有汽车制造完成！");
            }
        };

        //组装pipeline
        //    publisher->  door->wheel->seat消费者
        installDoor.subscribe(installWheel);
        installWheel.subscribe(installSeat);
        // 发布者 ->处理器
        pub.subscribe(installDoor);

        System.out.println("程序启动，等待制造完成...");


    }
}
